package com.pangyangyang.elasticsearch.plugin.canal.collector;/*
 * Created by gangzz.
 *  2015/7/3
 */

import com.alibaba.otter.canal.client.CanalConnector;
import com.pangyangyang.elasticsearch.plugin.river.Build;
import com.pangyangyang.elasticsearch.plugin.util.Threads;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 *
 */
public class DefaultFailOverStrategy implements CanalCollectorFailOverStrategy {

    private int maxSleepSeconds = 30;
    private int failedTimes = 0;
    private static ESLogger logger = ESLoggerFactory.getLogger(Build.LOGGER_NAME);
    private CanalEventTask task;

    public DefaultFailOverStrategy(CanalEventTask task) {
        this.task = task;
    }

    @Override
    public boolean shouldStop() {
        return this.task.isStop();
    }

    @Override
    public void start(CanalConnector connector) {
        if (connector == null){
            throw new NullPointerException("canal collector is null!");
        }

        int currentFailedTimes = 0;
        while (!this.task.isStop()) {
            try {
                connector.connect();
                connector.subscribe();
                currentFailedTimes = 0;
                break;
            } catch (Throwable t) {
                failedTimes++;
                int sleepTime = Math.max(maxSleepSeconds, currentFailedTimes * 2);
                logger.warn("Canal connect failed, collector will retry after {} seconds.", sleepTime);
                logger.warn("Canal collector connect has retry {} times, total failed {} times from plugin started up.", currentFailedTimes, getFailedTimes());
                Threads.sleep(sleepTime, TimeUnit.SECONDS);
            } finally {
                stop(connector);
            }
        }
    }

    @Override
    public void stop(CanalConnector connector) {
        if (connector == null){
            return;
        }
        try{
            connector.unsubscribe();
        } catch (Throwable t){
            logger.debug(t.getMessage(), t);
        }
        try{
            connector.disconnect();
        } catch (Throwable t){
            logger.debug(t.getMessage(), t);
        }

    }

    @Override
    public void whenError(CanalConnector connector, Throwable t) {
        failedTimes ++;
    }

    public int getFailedTimes() {
        return failedTimes;
    }
}
